package net.qiqbframework.eventhandling;


import net.qiqbframework.common.Registration;
import net.qiqbframework.common.transaction.TransactionManager;
import net.qiqbframework.lifecycle.Lifecycle;
import net.qiqbframework.lifecycle.LifecycleHandlerRegister;
import net.qiqbframework.lifecycle.Phase;
import net.qiqbframework.messaging.SubscribableMessageSource;
import net.qiqbframework.messaging.task.DefaultMessageTask;
import net.qiqbframework.messaging.task.MessageTask;

import javax.annotation.Nonnull;


public class SubscribingEventProcessor extends AbstractEventProcessor implements Lifecycle {

    private final SubscribableMessageSource<EventMessage<?>> messageSource;
    private final EventProcessingStrategy processingStrategy;
    private final TransactionManager transactionManager;

    private volatile Registration eventBusRegistration;


    protected SubscribingEventProcessor(SubscribingEventProcessorBuilder builder) {
        super(builder);
        this.messageSource = builder.getMessageSource();
        this.processingStrategy = builder.getProcessingStrategy();
        this.transactionManager = builder.getTransactionManager();
    }



    @Override
    public void registerLifecycleHandler(@Nonnull LifecycleHandlerRegister handle) {
        handle.onStart(Phase.LOCAL_MESSAGE_HANDLER_REGISTRATIONS, this::start);
        handle.onShutdown(Phase.LOCAL_MESSAGE_HANDLER_REGISTRATIONS, this::shutDown);
    }


    @Override
    public void start() {
        if (eventBusRegistration != null) {
            // This event processor has already been started
            return;
        }
        eventBusRegistration = messageSource.subscribe(eventMessage -> processingStrategy.handle(eventMessage, this::process));
    }

    @Override
    public boolean isRunning() {
        return eventBusRegistration != null;
    }

    @Override
    public boolean isError() {
        // this implementation will never stop because of an error
        return false;
    }


    protected void process(EventMessage<?> eventMessages) {

        try {
            MessageTask<? extends EventMessage<?>> unitOfWork = new DefaultMessageTask<>(eventMessages);
            unitOfWork.attachTransaction(transactionManager);
            unitOfWork.onPrepareCommit(this::prepareCommit);
            processInUnitOfWork(eventMessages, unitOfWork);
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw new EventProcessingException("Exception occurred while processing events", e);
        }
    }

    private void prepareCommit(MessageTask<? extends EventMessage<?>> messageWork) {
        //

    }

    @Override
    public void shutDown() {
        if (eventBusRegistration != null) {
            eventBusRegistration.cancel();
        }
        eventBusRegistration = null;
    }

    public SubscribableMessageSource<? extends EventMessage<?>> getMessageSource() {
        return messageSource;
    }

}
